package com.atguigu.flink.datastreamapi.sink;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * Created by Smexy on 2023/4/8
 *
 *  oldapi:  env.addSink(SinkFunction x)
 *              实现简单
 *  newapi:  env.sinkTo(Sink x)
 */
public class Demo6_CustomSink
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

                env
                   .socketTextStream("hadoop102", 8888)
                   .map(new WaterSensorMapFunction())
                   /*.addSink(new SinkFunction<WaterSensor>()
                   {
                       //每来一条数据，执行一次，负责把数据输出
                       @Override
                       public void invoke(WaterSensor value, Context context) throws Exception {

                           //如果是s1的传感器，使用 err输出，否则是否 out输出
                           if ("s1".equals(value.getId())){
                               System.err.println(value);
                           }else{
                               System.out.println(value);
                           }

                       }
                   })*/
                   //  RichSinkFunction<IN> extends AbstractRichFunction  implements SinkFunction
                   // 既有生命周期方法，还有sink的功能
                   .addSink(new RichSinkFunction<WaterSensor>()
                   {
                       @Override
                       public void close() throws Exception {
                           System.out.println("task创建时，释放连接");
                       }

                       @Override
                       public void open(Configuration parameters) throws Exception {
                           System.out.println("task创建时，获取连接");
                       }

                       @Override
                       public void invoke(WaterSensor value, Context context) throws Exception {

                           //如果是s1的传感器，使用 err输出，否则是否 out输出
                           if ("s1".equals(value.getId())){
                               System.err.println(value);
                           }else{
                               System.out.println(value);
                           }
                       }
                   })
                   ;

                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }


    }
}
